package com.o2o.cleaning.month.platform.ebusiness_plat.kuaishou

import java.time.LocalDate

import com.alibaba.fastjson.JSON
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.elasticsearch.spark._

object back_03_OBS_TO_ES {
  private val time: LocalDate = LocalDate.now()

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("Kuaishou_es")
      .master("local[*]")
      .config("es.nodes", "192.168.1.29")
      .config("es.port", "9200")
      .config("es.net.http.auth.user", "elastic")
      .config("es.net.http.auth.pass", "changeme")
      .config("es.batch.write.retry.count", "10")
      .config("es.batch.write.retry.wait", "60")
      .config("es.http.timeout", "100s")
      .config("cluster.name", "O2OElastic")
      .getOrCreate()
    val sc = spark.sparkContext

    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")

    /** *
      * 需要修改的参数信息
      * 处理完手动放  o2o-dataproces-group/zyf/livestreaming/kuaishou/es/${year}/${month}/  到  dws-data/g_data/2021/9/kuaishou/
      */
    val year = 2021
    val month = 12
    val platform = "kuaishou"
    val collection = if (month < 10) s"kuaishou_webcast_shop_list_210${month}" else s"kuaishou_webcast_shop_list_21${month}"
    val index = s"${year}_${platform}/${platform}_${year}_${month}"
    println(" index ---->" + index)
    //good路径
    val goodPath = s"s3a://o2o-dataproces-group/zyf/livestreaming/kuaishou/es/${year}/${month}/"
    println(" index ---->" + goodPath)

    //入库
    val df = spark.read.orc(goodPath)
      //      .where("subplatformName = 'xiaodian'")
      //    df.printSchema()
      //    df.show(false)
      //    .where("sellCount > 0 and salesAmount > 0")
      .toJSON.rdd.map(line => {

      //flavors类型不一致
      val lines = JSON.parseObject(line)

      val salesAmount: Double = lines.getDouble("salesAmount")
      val priceText: Double = lines.getOrDefault("price","0").toString.toDouble

      val sales: Double = salesAmount.formatted("%.2f").toDouble
      val price: Double = priceText.formatted("%.2f").toDouble

      lines.remove("starts")
      lines.put("salesAmount", sales)
      lines.put("priceText", price)
      lines.put("subplatformName", "来自小店")
      lines.remove("price")

      lines
    })
      .saveToEs(index, Map("es.mapping.id" -> "good_id"))
  }

}
